use crate::app::server::handle;
use crate::app::server::message;
use crate::app::server::message::Message;
use std::any::Any;
use std::io;
use std::net::SocketAddr;
use std::sync::mpsc::{channel, Receiver, Sender};
use std::sync::{Arc, RwLock};
use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
use tokio::net::TcpStream;
use tokio::prelude::io::*;

pub struct ConnEntity {
    //pub sock: TcpStream,
    pub addr: SocketAddr,
    pub handers: Arc<RwLock<handle::HandlerManage>>,
    pub sender: Sender<Message>,
    pub receiver: Receiver<Message>,
    pub hook: Arc<dyn handle::ConnHook + Send + 'static>,
    pub object: Box<dyn Any>,
}

unsafe impl Send for ConnEntity {}

impl ConnEntity {
    pub fn new(
        s: std::net::TcpStream,
        a: SocketAddr,
        h: Arc<RwLock<handle::HandlerManage>>,
        k: Arc<dyn handle::ConnHook + Send + 'static>,
    ) -> io::Result<ConnEntity> {
        let tsock = match TcpStream::from_std(s) {
            Err(e) => return Err(e),
            Ok(o) => o,
        };
        let (rs, ws) = tsock.into_split();
        Ok(ConnEntity {
            //sock: tsock,
            addr: a,
            handers: h,
            sender: Self::start_sender_thread(ws),
            receiver: Self::start_receiver_thread(rs),
            hook: k,
            object: Box::new(1),
        })
    }
    pub fn set_hook(mut self, hook: Arc<dyn handle::ConnHook + Send + 'static>) -> Self {
        self.hook = hook;
        self
    }

    pub async fn run(&mut self) -> io::Result<()> {
        // let my_hook = match hook.lock() {
        //     Err(e)=>{
        //         log_error_ln!("Could not get a fetch hook error {}",e.to_string());
        //         return Ok(());
        //     }
        //     Ok(o)=>{o}
        // };
        let hook = Arc::clone(&self.hook);
        hook.conn_success(self);
        self.worker();
        hook.conn_close(self);
        Ok(())
    }
    // pub fn close(&self) -> io::Result<()> {
    //     self.sock.shutdown(Shutdown::Both)
    // }
    pub fn send_vec(&self, header: u16, body: Vec<u8>) -> std::result::Result<(), String> {
        match self.sender.send(message::Message::new(header, body)) {
            Err(e) => Err(e.to_string()),
            Ok(_) => Ok(()),
        }
    }
    #[allow(unused_must_use)]
    pub fn start_sender_thread(mut sock: OwnedWriteHalf) -> Sender<Message> {
        let (sx, rx) = channel::<Message>();
        tokio::spawn(async move {
            // let mut sock =
            //     if let Ok(o) = socket.lock(){o}else{return;};
            loop {
                let recv_res = rx.recv();
                match recv_res {
                    Err(_e) => {
                        // log_error_ln!("start_sender_thread 接受信息错误{}", e.to_string());
                        break;
                    }
                    Ok(mut o) => {
                        if let Err(e) = sock.write_all(o.encode_to_slice().as_slice()).await {
                            log_error_ln!("发送信息错误:{}", e.to_string());
                            break;
                        }
                    }
                }
            }
            sock.shutdown().await;
        });
        sx
    }
    pub fn start_receiver_thread(mut sock: OwnedReadHalf) -> Receiver<Message> {
        let (sx, rx) = channel::<Message>();
        tokio::spawn(async move {
            loop {
                let mut header_buf = [0; 4];
                let (status, len) = match sock.read_exact(&mut header_buf).await {
                    Err(e) => {
                        let msg = message::READ_OVER.clone().set_error(e.to_string());
                        if let Err(_e) = sx.send(msg) {}
                        break;
                    }
                    Ok(_n) => (
                        u16::from_le_bytes([header_buf[0], header_buf[1]]),
                        u16::from_le_bytes([header_buf[2], header_buf[3]]),
                    ),
                };
                let mut body_buf = vec![0u8; len as usize];
                match sock.read_exact(&mut body_buf).await {
                    Err(e) => {
                        if let Err(e) = sx.send(message::READ_OVER.clone().set_error(e.to_string()))
                        {
                            log_error_ln!(
                                "start_receiver_thread 向客户端发送数据错误:{}",
                                e.to_string()
                            );
                            break;
                        }
                    }
                    Ok(_) => {}
                }
                if let Err(e) = sx.send(message::Message::new(status, body_buf)) {
                    log_error_ln!("start_receiver_thread 发送信息错误:{}", e.to_string());
                    break;
                }
            }
        });
        rx
    }
    pub fn worker(&mut self) {
        loop {
            let msg = match self.receiver.recv() {
                Err(_e) => {
                    // log_error_ln!("worker 接受信息出错:{}", e.to_string());
                    break;
                }
                Ok(o) => o,
            };
            let handle_lock = self.handers.clone();
            let handle = match handle_lock.read() {
                Err(e) => {
                    log_error_ln!("worker 获取锁错误:{}", e.to_string());
                    continue;
                }
                Ok(o) => o,
            };
            if let Some(m) = handle.get_call_func(msg.header) {
                let (ok, remsg) = m.handle(self, msg);
                match remsg.status {
                    message::MessageStatus::Over => break,
                    message::MessageStatus::Normal => {
                        if remsg.header == 0 {
                            continue;
                        }
                        if let Err(e) = self.sender.send(remsg) {
                            log_error_ln!("{}", e.to_string())
                        }
                    } // _=>{}
                }
                if !ok {
                    break;
                }
            } else {
                if let Err(e) = self.sender.send(message::MSG_UNDEFINED.clone()) {
                    log_error_ln!("{}", e.to_string())
                }
            }
        }
    }
}
